package cn.dmp.report

import cn.dmp.beans.Log
import cn.dmp.utils.RptUItils
import org.apache.commons.lang.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

/**
  * 需求：媒体报表分析
  *     broadcast 广播变量的使用（将字典文件广播到Executor端）
  * 假设一个场景：如果字典文件一直在变，那Executor端如何获取该字典文件数据呢？（解决方案：如果Executor端要匹配Driver端的一直变化的字典文件数据，可以用redis数据库，将字典文件数据放到redis里面，以后需Executor要匹配字典数据的话可以直接读取redis数据库，并且redis读取的速度挺快的）
  * 广播变量的值能改变吗？一旦广播出去就不能改了。
  */

// extends App 就不用写main函数了。
object AppDictAnalyse extends App {

  // 0.判断参数
  if (args.length != 3){
    println(
      """
        |cn.dmp.report.ProCityRpt_Version3
        |参数：
        |   InputPath
        |   dictFilePath
        |   resultOutputPath
      """.stripMargin)
  }

  // 1.接收程序参数
  val Array(inputPath, dictFilePath, resultOutputPath) = args

  // 2.创建SparkSession
  val sparkSession: SparkSession = SparkSession.builder().appName("ProCityRpt_Version3").master("local[*]")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //spark优化：spark.serializer修改磁盘序列化方式
    .enableHiveSupport()
    .getOrCreate()

  // 3.读取字典文件，将字典文件数据处理后进行广播到Executor端
  /**
    * 注意：
    * (1)collect()函数是为了将每个分片的数据都收集起来，如果不收集的话那么这里返回的可能是个分片的部分数据！！！ .toMap是将其变成map形式的数据
    * (2)更不能在map外面定义一个hashMap,然后在map里面不停的加入数据给hashMap，然后最终以为hashMap收集到的是所有的值。这是错误的（真实情况是hashMap谁也没收到，为空）！！！！
    *       因为在map外面定义的hashMap是在Driver端而map里面是在Executor端！！！！,
    *
    */
  val dictDataSet: Dataset[String] = sparkSession.read.textFile(dictFilePath)
  val dictMap: Map[String, String] = dictDataSet.map(line => {
    val fields = line.split("\t", -1)
    (fields(4), fields(1))
  }).collect().toMap

  // 将字典数据广播到Executor端
  val broadCast: Broadcast[Map[String, String]] = sparkSession.sparkContext.broadcast(dictMap)

  // 读取日志数据
  val res: RDD[String] = sparkSession.read.textFile(inputPath)
    .map(line => line.split(",", -1))
    .filter(_.length >= 85)
    .map(Log(_)).filter(log => !log.Appid.isEmpty || !log.Appname.isEmpty)
    .map(log => {
      var newAppName = log.Appname
      if (!StringUtils.isNotEmpty(newAppName)) {
        newAppName = broadCast.value.getOrElse(log.Appid, "未知")
      }

      val req: List[Double] = RptUItils.caculateReq(log.Requestmode, log.Processnode)
      val rtb: List[Double] = RptUItils.caculateRtb(log.Iseffective, log.Isbilling, log.Isbid, log.Adorderid, log.Iswin, log.Winprice, log.Adpayment)
      val showClick: List[Double] = RptUItils.caculateShowClick(log.Requestmode, log.Iseffective)

      (newAppName, req ++ rtb ++ showClick)
    }).rdd.reduceByKey((list1, list2) => {
    list1.zip(list2).map(t => t._1 + t._2)
  }).map(t => t._1 + "," + t._2.mkString(","))

  res.saveAsTextFile(resultOutputPath)

  sparkSession.stop()


  /**
    * 输入目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\data\2016-10-01_06_p1_invalid.1475274123982.log.FINISH
    * 输入目录：字典文件数据
    * 输出目录：E:\大数据培训视频\项目+spark面试+flink+项目\项目\DMP\AppDictAnalyse
    */




}
